/*-------------------------------------------------------------------------
 *
 * tsm_system_time.c
 *	  支持 SYSTEM_TIME tablesample 方法的例程
 *
 * 此处的需求是生成尽可能多的随机样本，
 * 而不超过指定的时间。我们使用块采样方法。为了确保如果需要
 * 将访问整个关系，我们从一个随机选择的块开始，然后
 * 以随机选择的步长前进，但此步长与关系的 nblocks 相对互质。
 *
 * 由于时间依赖性，此方法必然不可重复。
 * 然而，我们尽可能减少令人惊讶的行为，通过每个查询仅选择一次采样模式，类似于 tsm_system_rows。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  contrib/tsm_system_time/tsm_system_time.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include <math.h>

#include "access/relscan.h"
#include "access/tsmapi.h"
#include "catalog/pg_type.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "utils/sampling.h"
#include "utils/spccache.h"

PG_MODULE_MAGIC;

PG_FUNCTION_INFO_V1(tsm_system_time_handler);


/* 私有状态 */
typedef struct
{
	uint32		seed;			/* 随机种子 */
	double		millis;			/* 采样的时间限制 */
	instr_time	start_time;		/* 扫描开始时间 */
	OffsetNumber lt;			/* 从当前块返回的最后一个元组 */
	BlockNumber doneblocks;		/* 已扫描块的数量 */
	BlockNumber lb;				/* 最后访问的块 */
	/* 这三个值在重新扫描期间不变: */
	BlockNumber nblocks;		/* 关系中的块数量 */
	BlockNumber firstblock;		/* 采样的第一个块 */
	BlockNumber step;			/* 步长，若尚未设置则为0 */
} SystemTimeSamplerData;

static void fc_system_time_samplescangetsamplesize(PlannerInfo *fc_root,
												RelOptInfo *fc_baserel,
												List *fc_paramexprs,
												BlockNumber *fc_pages,
												double *fc_tuples);
static void fc_system_time_initsamplescan(SampleScanState *fc_node,
									   int fc_eflags);
static void fc_system_time_beginsamplescan(SampleScanState *fc_node,
										Datum *fc_params,
										int fc_nparams,
										uint32 fc_seed);
static BlockNumber fc_system_time_nextsampleblock(SampleScanState *fc_node, BlockNumber fc_nblocks);
static OffsetNumber fc_system_time_nextsampletuple(SampleScanState *fc_node,
												BlockNumber fc_blockno,
												OffsetNumber fc_maxoffset);
static uint32 fc_random_relative_prime(uint32 fc_n, pg_prng_state *fc_randstate);


/*
 * 为SYSTEM_TIME方法创建TsmRoutine描述符。
 */
Datum tsm_system_time_handler(PG_FUNCTION_ARGS)
{
	TsmRoutine *fc_tsm = makeNode(TsmRoutine);

	fc_tsm->parameterTypes = list_make1_oid(FLOAT8OID);

	/* 请参阅文件开头的说明 */
	fc_tsm->repeatable_across_queries = false;
	fc_tsm->repeatable_across_scans = false;

	fc_tsm->SampleScanGetSampleSize = fc_system_time_samplescangetsamplesize;
	fc_tsm->InitSampleScan = fc_system_time_initsamplescan;
	fc_tsm->BeginSampleScan = fc_system_time_beginsamplescan;
	fc_tsm->NextSampleBlock = fc_system_time_nextsampleblock;
	fc_tsm->NextSampleTuple = fc_system_time_nextsampletuple;
	fc_tsm->EndSampleScan = NULL;

	PG_RETURN_POINTER(fc_tsm);
}

/*
 * 样本大小估计。
 */
static void fc_system_time_samplescangetsamplesize(PlannerInfo *fc_root,
									RelOptInfo *fc_baserel,
									List *fc_paramexprs,
									BlockNumber *fc_pages,
									double *fc_tuples)
{
	Node	   *fc_limitnode;
	double		fc_millis;
	double		fc_spc_random_page_cost;
	double		fc_npages;
	double		fc_ntuples;

	/* 尝试提取限制时间规格的估计值 */
	fc_limitnode = (Node *) linitial(fc_paramexprs);
	fc_limitnode = estimate_expression_value(fc_root, fc_limitnode);

	if (IsA(fc_limitnode, Const) &&
		!((Const *) fc_limitnode)->constisnull)
	{
		fc_millis = DatumGetFloat8(((Const *) fc_limitnode)->constvalue);
		if (fc_millis < 0 || isnan(fc_millis))
		{
			/* 如果值无效则使用默认毫秒 */
			fc_millis = 1000;
		}
	}
	else
	{
		/* 如果未获得非空的Const，则使用默认毫秒 */
		fc_millis = 1000;
	}

	/* 获取规划者对于每页读取的成本想法 */
	get_tablespace_page_costs(fc_baserel->reltablespace,
							  &fc_spc_random_page_cost,
							  NULL);

	/*
	 * 通过假设成本数字以毫秒为单位来估计我们可以读取的页面数量。
	 * 这完全是错误的，但我们必须采取某种措施来生成估计，并且
	 * 没有更好的答案。
	 */
	if (fc_spc_random_page_cost > 0)
		fc_npages = fc_millis / fc_spc_random_page_cost;
	else
		fc_npages = fc_millis;		/* 更加荒谬，但你能怎么办呢？ */

	/* 限制为合理值 */
	fc_npages = clamp_row_est(Min((double) fc_baserel->pages, fc_npages));

	if (fc_baserel->tuples > 0 && fc_baserel->pages > 0)
	{
		/* 基于元组密度估计返回的元组数量 */
		double		fc_density = fc_baserel->tuples / (double) fc_baserel->pages;

		fc_ntuples = fc_npages * fc_density;
	}
	else
	{
		/* 由于缺乏数据，假设每页一个元组 */
		fc_ntuples = fc_npages;
	}

	/* 限制为估计的关系大小 */
	fc_ntuples = clamp_row_est(Min(fc_baserel->tuples, fc_ntuples));

	*fc_pages = fc_npages;
	*fc_tuples = fc_ntuples;
}

/*
 * 在执行器设置期间初始化。
 */
static void fc_system_time_initsamplescan(SampleScanState *fc_node, int fc_eflags)
{
	fc_node->tsm_state = palloc0(sizeof(SystemTimeSamplerData));
	/* 请注意，上述内容使得tsm_state->step等于零 */
}

/*
 * 检查参数并准备进行样本扫描。
 */
static void fc_system_time_beginsamplescan(SampleScanState *fc_node,
							Datum *fc_params,
							int fc_nparams,
							uint32 fc_seed)
{
	SystemTimeSamplerData *fc_sampler = (SystemTimeSamplerData *) fc_node->tsm_state;
	double		fc_millis = DatumGetFloat8(fc_params[0]);

	if (fc_millis < 0 || isnan(fc_millis))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_TABLESAMPLE_ARGUMENT),
				 errmsg("sample collection time must not be negative")));

	fc_sampler->seed = fc_seed;
	fc_sampler->millis = fc_millis;
	fc_sampler->lt = InvalidOffsetNumber;
	fc_sampler->doneblocks = 0;
	/* start_time，lb将在第一次NextSampleBlock调用期间初始化 */
	/* 我们故意不在此处更改nblocks/firstblock/step */
}

/*
 * 选择下一个要采样的块。
 *
 * 使用线性探测算法选择下一个块。
 */
static BlockNumber fc_system_time_nextsampleblock(SampleScanState *fc_node, BlockNumber fc_nblocks)
{
	SystemTimeSamplerData *fc_sampler = (SystemTimeSamplerData *) fc_node->tsm_state;
	instr_time	fc_cur_time;

	/* 扫描中的第一次调用? */
	if (fc_sampler->doneblocks == 0)
	{
		/* 查询中的第一次扫描? */
		if (fc_sampler->step == 0)
		{
			/* 现在初始化，既然我们有了扫描描述符 */
			pg_prng_state fc_randstate;

			/* 如果关系为空，则没有要扫描的内容 */
			if (fc_nblocks == 0)
				return InvalidBlockNumber;

			/* 在此设置步骤中我们只需要一个随机数生成器 */
			sampler_random_init_state(fc_sampler->seed, &fc_randstate);

			/* 每个查询仅需计算一次nblocks/firstblock/step */
			fc_sampler->nblocks = fc_nblocks;

			/* 在关系中选择随机起始块 */
			/* （实际上这是访问的第一个块的前任） */
			fc_sampler->firstblock = sampler_random_fract(&fc_randstate) *
				fc_sampler->nblocks;

			/* 找到相对于步长的相对素数以进行线性探测 */
			fc_sampler->step = fc_random_relative_prime(fc_sampler->nblocks, &fc_randstate);
		}

		/* 重新初始化lb和start_time */
		fc_sampler->lb = fc_sampler->firstblock;
		INSTR_TIME_SET_CURRENT(fc_sampler->start_time);
	}

	/* 如果我们已经读取了关系中的所有块，则完成 */
	if (++fc_sampler->doneblocks > fc_sampler->nblocks)
		return InvalidBlockNumber;

	/* 如果我们已用完所有分配的时间，则完成 */
	INSTR_TIME_SET_CURRENT(fc_cur_time);
	INSTR_TIME_SUBTRACT(fc_cur_time, fc_sampler->start_time);
	if (INSTR_TIME_GET_MILLISEC(fc_cur_time) >= fc_sampler->millis)
		return InvalidBlockNumber;

	/*
	 * 在查询中的扫描之间，scan->rs_nblocks降到下面的可能性很小；
	 * 但以防万一，循环直到选择的块号小于scan->rs_nblocks。
	 * 我们不关心自第一次扫描以来scan->rs_nblocks是否增加。
	 */
	do
	{
		/* 前进lb，使用uint64算术以防止溢出 */
		fc_sampler->lb = ((uint64) fc_sampler->lb + fc_sampler->step) % fc_sampler->nblocks;
	} while (fc_sampler->lb >= fc_nblocks);

	return fc_sampler->lb;
}

/*
 * 选择当前块中下一个采样的元组。
 *
 * 在块采样中，我们只想采样每个选定块中的所有元组。
 *
 * 当我们到达块的末尾时，返回InvalidOffsetNumber，这告诉
 * SampleScan转到下一个块。
 */
static OffsetNumber fc_system_time_nextsampletuple(SampleScanState *fc_node,
							BlockNumber fc_blockno,
							OffsetNumber fc_maxoffset)
{
	SystemTimeSamplerData *fc_sampler = (SystemTimeSamplerData *) fc_node->tsm_state;
	OffsetNumber fc_tupoffset = fc_sampler->lt;

	/* 向页面上的下一个可能偏移量推进 */
	if (fc_tupoffset == InvalidOffsetNumber)
		fc_tupoffset = FirstOffsetNumber;
	else
		fc_tupoffset++;

	/* 完成？ */
	if (fc_tupoffset > fc_maxoffset)
		fc_tupoffset = InvalidOffsetNumber;

	fc_sampler->lt = fc_tupoffset;

	return fc_tupoffset;
}

/*
 * 计算两个uint32的最大公约数。
 */
static uint32 fc_gcd(uint32 fc_a, uint32 fc_b)
{
	uint32		fc_c;

	while (fc_a != 0)
	{
		fc_c = fc_a;
		fc_a = fc_b % fc_a;
		fc_b = fc_c;
	}

	return fc_b;
}

/*
 * 如果可能，选择一个小于n且互为素数的随机值
 * （否则返回1）。
 */
static uint32 fc_random_relative_prime(uint32 fc_n, pg_prng_state *fc_randstate)
{
	uint32		fc_r;

	/* 安全检查以避免无限循环或对于小 n 的零结果。 */
	if (fc_n <= 1)
		return 1;

	/*
	 * 这应该只需要 2 或 3 次迭代，因为两个数互质的概率大约为 61%；但以防万一，我们将在循环中包含 CHECK_FOR_INTERRUPTS。
	 */
	do
	{
		CHECK_FOR_INTERRUPTS();
		fc_r = (uint32) (sampler_random_fract(fc_randstate) * fc_n);
	} while (fc_r == 0 || fc_gcd(fc_r, fc_n) > 1);

	return fc_r;
}
